package com.JadePenG.search.service.listener;

import com.JadePenG.gossip.pojo.News;
import com.JadePenG.search.service.impl.SolrIndexWrite;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;

/**
 * @author Peng
 * @Description 自定义监听类,消费新闻数据,调用索引写入方法
 */
@Component
public class SpiderKafkaConsumer implements MessageListener<String, String> {

    @Autowired
    private SolrIndexWrite solrIndexWriter;
    private Gson gson = new Gson();


    @Override
    public void onMessage(ConsumerRecord<String, String> data) {

        try {
            String newJsonString = data.value();

            News news = gson.fromJson(newJsonString, News.class);

            SimpleDateFormat format1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            SimpleDateFormat format2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
            String time = news.getTime();
            Date oldTime = format1.parse(time);
            String newTime = format2.format(oldTime);
            news.setTime(newTime);

            //调用索引写入方法, 将新闻数据保存到solr中
            solrIndexWriter.saveBeans(Arrays.asList(news));

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
